/// 包装channel
use futures_channel::mpsc::{unbounded, UnboundedReceiver, UnboundedSender};
use crate::worker::envelop::{Envelop, ToEnvelop};
use crate::{Worker, Message, Handler};
use futures_util::stream::{FusedStream, Stream};
use std::pin::Pin;
use std::task::{Poll, Context};
use crate::worker::envelop::EnvelopProxy;

pub struct Sender<W: Worker> {
    inner: UnboundedSender<Envelop<W>>
}

impl<W: Worker> Sender<W>
{
    pub fn send<M>(&self, msg: M)
        where W: Handler<M>,
              M: Message + Send + 'static,
              W::Context: ToEnvelop<W, M>
    {
        self.inner.unbounded_send(<W::Context as ToEnvelop<W, M>>::pack(msg)).expect("send error");
    }
}

impl<W: Worker> Clone for Sender<W> {
    fn clone(&self) -> Self {
        Sender {
            inner: self.inner.clone()
        }
    }
}

pub struct Receiver<W: Worker> {
    inner: UnboundedReceiver<Envelop<W>>
}

impl<W: Worker> Receiver<W> {
    pub fn is_terminated(&self) -> bool {
        self.inner.is_terminated()
    }

    pub fn poll(&mut self, cx: &mut Context<'_>, act: &mut W) {
        loop {
            match Pin::new(&mut self.inner).poll_next(cx) {
                // 无消息处理时让出处理器
                Poll::Pending => {
                    break;
                }
                // 当有消息，一直处理消息即可
                Poll::Ready(Some(mut envelop)) => {
                    envelop.handle(act);
                }
                // 不会返回空消息
                Poll::Ready(None) => unreachable!(),
            }
        }
    }
}


pub fn channel<W: Worker>() -> (Sender<W>, Receiver<W>) {
    let (sed, rev) = unbounded();
    (
        Sender {
            inner: sed
        },
        Receiver {
            inner: rev
        }
    )
}
